Hive

[Hive] Hive之UDF函数

自定义UDF、自定义临时函数和永久函数

Posted by 李玉坤 on 2017-10-23

UDF：普通的用户自定义函数。接受单行输入，并产生单行输出。如转换字符串大小写，获取字符串长度等。
UDAF：用户定义聚集函数（User-defined aggregate function）。接受多行输入，并产生单行输出。比如MAX，COUNT函数。比如sum/min
UDTF：用户定义表生成函数（User-defined table-generating function）。接受单行输入，并产生多行输出（即一个表），不是特别常用。比如：lateral view explode()

编写测试UDF函数

pom文件添加

1
2
3
4
5
6
7
<hive.version>1.1.0-cdh5.7.0</hive.version>

  <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-exec</artifactId>
      <version>${hive.version}</version>
    </dependency>

UDF 数据拼接函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import org.apache.hadoop.hive.ql.exec.UDF;

public class HelloUDF extends UDF{

    public String evaluate(String input){
        //TODO:工作中的详细业务功能
        return "Hello："+input;
    }

    public String evaluate(String input1,String input2){
        //TODO:工作中的详细业务功能
        return "Hello："+input1+"=>"+input2;
    }


}

AVG udf函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.text.DecimalFormat;

public class AvgScore extends GenericUDF {
    private static final String FUNC_NAME="AVG_SCORE";
    private transient MapObjectInspector mapOi;

    DecimalFormat df = new DecimalFormat("#.##");
    @Override
    public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
        //检测函数参数个数
        //检测函数参数类型
        mapOi = (MapObjectInspector) objectInspectors[0];
        return PrimitiveObjectInspectorFactory.javaDoubleObjectInspector;
    }

    @Override
    public Object evaluate(DeferredObject[] deferredObjects) throws HiveException {

        Object o = deferredObjects[0].get();
        double v = mapOi.getMap(o).values().stream().mapToDouble(a -> Double.parseDouble(a.toString())).average().orElse(0.0);
        return Double.parseDouble(df.format(v));
    }

    @Override
    public String getDisplayString(String[] strings) {
        return "func(map)";
    }
}

打包上传到服务器 

1
2
3
4
5
[hadoop@hadoop hive]$ pwd
/home/hadoop/data/hive
[hadoop@hadoop hive]$ ll
总用量 8
-rw-r--r--. 1 hadoop hadoop 7483 4月  18 10:39 hive_UDF.jar

Hive中创建函数

创建临时函数

临时函数仅对当前session(黑窗口)有效

添加jar包到hive
hive> add jar /home/hadoop/data/hive/hive_UDF.jar;

创建临时函数
语法：

1
2
3
CREATE TEMPORARY FUNCTION function_name AS class_name;   
                    function_name函数名    
                    class_name 类路径，包名+类名

实例：
创建sayhello函数 

1
2
3
4
5
6
7
8
9
10
11
hive> create temporary function sayhello as 'com.kun.hive.HelloUDF';
OK
Time taken: 0.485 seconds
hive>
hive> show functions; 【查看可以看到sayhello】
OK 
rpad
rtrim
sayhello
second
sentences

测试：

1
2
3
4
5
6
7
8
9
hive> select sayhello('wuwang');
OK
Hello：wuwang
Time taken: 2.712 seconds, Fetched: 1 row(s)
hive> select sayhello('wuwang','nihao');
OK
Hello：wuwang=>nihao
Time taken: 0.128 seconds, Fetched: 1 row(s)
hive>

删除临时函数 ：

语法：

DROP TEMPORARY FUNCTION [IF EXISTS] function_name;

实例测试： 

1
2
3
4
5
6
hive> DROP TEMPORARY FUNCTION IF EXISTS sayhello;
OK
Time taken: 0.003 seconds
hive> select sayhello('wuwang');
FAILED: SemanticException [Error 10011]: Line 1:7 Invalid function 'sayhello'
hive>

创建永久函数

上传语法：

1
2
3
CREATE FUNCTION [db_name.]function_name AS class_name 
            [USING JAR|FILE|ARCHIVE 'file_uri' [, JAR|FILE|ARCHIVE 'file_uri'] ]; 
            file_uri:是hdfs上的jar包目录

上传实例：

1
2
3
4
5
6
[hadoop@hadoop hive-1.1.0-cdh5.7.0]$ hadoop fs -mkdir /lib
[hadoop@hadoop hive-1.1.0-cdh5.7.0]$ hadoop fs -put /home/hadoop/data/hive/hive_UDF.jar /lib/
[hadoop@hadoop hive-1.1.0-cdh5.7.0]$ hadoop fs -ls /lib/
Found 1 items
-rw-r--r--   1 hadoop supergroup       7483 2019-04-18 11:10 /lib/hive_UDF.jar
[hadoop@hadoop hive-1.1.0-cdh5.7.0]$

创建语法：

1
2
3
4
CREATE TEMPORARY FUNCTION function_name AS class_name USING JAR path;   
                    function_name函数名    
                    class_name 类路径，包名+类名 
                    path jar包hdfs路径

创建实例：

1
2
3
4
5
6
7
8
9
10
11
12
hive> CREATE FUNCTION sayhello2 AS 'com.kun.hive.HelloUDF' USING JAR 'hdfs://hadoop:8020/lib/hive_UDF.jar'
    > ;
converting to local hdfs://hadoop:8020/lib/hive_UDF.jar
Added [/tmp/4b56b71a-d406-4ee7-89db-9d3c97a19e81_resources/hive_UDF.jar] to class path
Added resources: [hdfs://hadoop:8020/lib/hive_UDF.jar]
OK
Time taken: 0.552 seconds
hive> show functions;【查看】
OK
dayofmonth
decode
default.sayhello2

测试：

1
2
3
4
hive> select sayhello2('wuwang');
OK
Hello：wuwang
Time taken: 1.228 seconds, Fetched: 1 row(s)

另外开一个session测试：

可以查看到元数据：

重新启动hadoop和hive发现sayhello2依旧可用

1
2
3
4
5
6
7
8
hive> select sayhello2('wu');
converting to local hdfs://hadoop:8020/lib/hive_UDF.jar
Added [/tmp/9dcbba84-d9db-4bcf-8eed-2cf20abfc510_resources/hive_UDF.jar] to class path
Added resources: [hdfs://hadoop:8020/lib/hive_UDF.jar]
OK
Hello：wu
Time taken: 2.477 seconds, Fetched: 1 row(s)
hive>

Hive直接使用不用创建函数

前置java版本为1.8

*1、下载源码 *
hive-1.1.0-cdh5.7.0-src.tar.gz
http://archive.cloudera.com/cdh5/cdh/5/hive-1.1.0-cdh5.7.0-src.tar.gz

2、解压源码
tar -zxvf hive-1.1.0-cdh5.7.0-src.tar.gz -C

3、将HelloUDF.java文件增加到HIVE源码中

[root@hadoop01 hive-1.1.0-cdh5.7.0]# cp HelloUDF.java 到 ql/src/java/org/apache/hadoop/hive/ql/udf/
更改HelloUDF.java的第一行包路径为package org.apache.hadoop.hive.ql.udf;

4、修改FunctionRegistry.java 文件
[root@hadoop01 hive-1.1.0-cdh5.7.0]# cd ql/src/java/org/apache/hadoop/hive/ql/exec/
vi FunctionRegistry.java
在import中增加：import org.apache.hadoop.hive.ql.udf.HelloUDF;
在代码体头部 static 块中添加：system.registerUDF(“helloUDF”, HelloUDF.class, false);

5、重新编译
[root@hadoop01 hive-1.1.0-cdh5.7.0]# mvn clean package -DskipTests -Phadoop-2 -Pdist

6、两种配置方法

  1. 把编译后的jar包作为配置包

  2. 将编译后带UDF函数的包复制到旧hive环境

    到
    [root@hadoop01 target]# pwd /root/hive-1.1.0-cdh5.7.0/packaging/target/apache-hive-1.1.0-cdh5.7.0-bin/apache-hive-1.1.0-cdh5.7.0-bin/lib
    下，找到hive-exec-1.1.0-cdh5.7.0.jar包，将此jar包拷贝到旧hive环境/lib目录下
    命令：
    最后启动hive

8、测试：
hive
hive (default)> show functions ; – 能查看到有 helloudf
hive> select helloudf(“wuwang”,”nihao”);
OK
Hello：wuwang=>nihao
Time taken: 2.086 seconds, Fetched: 1 row(s)
hive>




FEATURED TAGS
Hive